企查查基于 Apache Iceberg 与 Arctic 构建实时湖仓实践
Arctic 是一个开放式架构下的湖仓管理系统，在开放的数据湖格式之上， 提供更多面向流和更新场景的优化，以及一套可插拔的数据自优化机制和管理服务。
作者介绍
仲启尚，企查查大数据架构部，资深数据组件工程师，目前主要参与 Flink 实时计算和基于 Apache Iceberg 的湖仓一体方向的研发，Apache Flink & Arctic Contributor。
摘要
本文主要介绍了企查查基于 Apache Iceberg+Arctic[1] 在的实时湖仓的落地与实践，我们积累出了一些经验希望通过本篇文章和大家分享。
主要围绕以下几个方面展开：
背景及业务痛点
遇见 Apache Iceberg
遇见 Arctic
在企查查的落地与实践
未来规划
背景及业务痛点
在企查查，数据来源主要包括云上环境和 IDC 机房的各种数据源，数据源包括 MySQL，TiDB 集群，MongoDB 集群和日志数据。数据被收集到 Kafka 集群，分别用于主要基于 Spark 引擎的离线计算和基于 Flink 引擎的实时计算。计算结果通过同步组件再同步到不同的数据库引擎中，为不同的产品提供查询服务。
ODS 入仓流程如下：
随着公司业务发展、用户数量持续增长以及数仓建设不断完善，元数据种类的增长，痛点也越发明显。主要包括以下的几个方面：
第一，稳定性。抽取数据时间集中，对数据库压力较大。数据源的全量入仓场景，对源库的批量拉取，造成了数据库非常不稳定的问题。
第二，时效性。离线 T+1，增量合并也只能达到小时级别，而且链路长，维护成本高。
第三，业务赋能。不允许业务直接对线上的 TiDB、MongoDB 数据库的数据进行查询，会影响到集群稳定，但是业务方（产品、测试）有需要对线上的数据进行即席查询。
遇见 Apache Iceberg
开源数据湖项目主要以 Delta Lake、Hudi、Iceberg 为代表，这三者各自提供了一种数据湖的Table Format，三者各有各的背景和特点。
Delte Lake 主要绑定 Spark 引擎，企查查内部实时计算主要以 Flink 引擎为主，在长期发展上不匹配，从而导致我们第一个排查 Delta Lake。（2.0开源之后，社区也在积极拥抱 Flink 计算引擎)
主要对 Hudi 和 Iceberg 两者进行了调研，在查看了一些对比的案例[2]和内部测试之后，以下几个是我们最终选择 Iceberg 的原因（仅针对调研时两者的状态进行，不代表当前现状）：
Iceberg Table format 更加开放，设计上面做了很好的抽象，没有强绑定某一个特定的引擎，更容易对接内部多个计算引擎 Flink、Spark、Trino。
V2 版本对于 CDC 场景的接入有比较完善的支持了。
Iceberg 已准备进行 1.0.0 的 release。
Hudi 查询表名后面指定后缀_rt / _cow，还会有一些默认的字段，我们更希望对 hive 有更好的兼容性的 Table format，从而减少用户使用成本。
引擎侧 ORC 格式的支持进度 Hudi 是慢于 Iceberg，内部主要为 ORC 存储。
遇见 Arctic
在 CDC 数据实时同步入湖场景中，CDC 数据包含有大量的删除更新操作，在 Iceberg 中会生成 Delete 文件，这些文件告诉读取方要跳过的已被替换或删除的行，读取时要合并 Data 文件和 Delete 文件。Flink Job 每次 Commit 都会带来小文件，随着数据持续写入到 Iceberg 表中，数据的查询效率会逐渐降低，因为打开文件所需的处理时间会增加。为了保证 Iceberg 表的查询性能，对数据的合并提出了挑战。
调研发现有多种方式来进行数据合并：
采用和 Hudi 类似的 Inline compaction方式，较为方便，但是也会带来资源使用和稳定性上面的问题，Hudi 社区也是推荐一个外部服务[3]来做合并。
周期调度执行合并小文件，像 Iceberg 提供的基于 Spark 名为 rewrite_data_files 的 procedures，但是需要开发一套触发机制，而且执行的代价比较大。
通过建立一个外部服务，通过轮询元数据的状态来触发文件的合并，过期文件的清理。
最终调研了各大厂数据湖落地的情况[4][5][6]，一个持续稳定的独立合并服务可以更好地对资源进行管控和利用且不影响写入任务，是使用好 Iceberg V2 表的基础，当我们内部还在讨论怎么落地这个外部服务的阶段，Arctic 开源了，同时首先对 Native Iceberg Table Format 的小文件合并服务进行了支持，同时还包括了孤儿文件清理，过期快照清理服务。
一切都是刚刚好的遇见。
在企查查的落地与实践
针对业务的痛点，企查查在 Hadoop 生态上引入数据湖组件，启动实时湖仓项目。通过一整套周边服务和平台化，打造一套高性能的湖仓体验。
下图为湖仓一体架构设计：
数据源主要来自 TiDB、MySQL、MongoDB 等业务数据库的 Changelog，存储现在主要是在 HDFS，后续将会接入对象存储来降低存储的成本。数仓体系现在主要是以 Hive 为主，将会逐渐过渡到 Iceberg Table Format 两种共存的现状，通过 Arctic 来治理 Iceberg 实时写入的文件问题。最上层计算引擎主要是 Flink、Spark、Trino 三种为主，Flink 主要是数据接入，Spark 来执行 / 替换现有的数据分层的数据处理；即席查询主要以 Trino 为主，也可以使用 Kyuubi + Spark。
数据入湖
MySQL 通过 Flink CDC 消费 binlog 同步数据，由于每张表都需要启动一个 Flink CDC 任务来进行同步，当任务较多时，对数据库产生较大的压力，每个任务都会拉取全量的 binlog，在 Flink 任务进行过滤；某些表可能会被多个任务使用；还伴随着的是跨云专线流量的占用也会放大。
TiDB 通过 TiCDC 同步，以 Canal 格式推送到 Kafka 中。Flink CDC 其实是支持 TiDB 数据源的，一方面 Flink CDC 使用的 tivk-client 需要使用到 TiKV 的权限，直连到每个 TiKV 去获取数据，需要的权限比较大；还有一方面是内部 TiCDC 使用经验和运维体系成熟，稳定性有保障，所以最终没有选择 Flink CDC 去同步 TiDB。
基于以上背景，MySQL 通过库级别的 Flink CDC 任务来同步到 Kafka，减少对 MySQL 备库的影响和专线流量的使用，TiDB 通过社区提供的 TiCDC 集群同步到 Kafka。统一 MySQL 和 TiDB 实时同步的后续数据链路，通过消息队列 Kafka 进行数据分发。通过 Flink Hybrid Source 自定义 Connector 来接入全量（MySQL / TiDB） + 增量（Kafka ）实现数据同步，支持 Initial / Batch / Stream 三种同步模式。近实时地写 Iceberg 表，默认 5 分钟一个 Checkpoint，即数据可见性延迟最多为5分钟。可以根据业务实时性要求来调整，最小间隔限制为 10 s，避免产生过多小文件给小文件合并带来较大压力。
入湖流程如下图所示：
湖仓管理
Arctic 是一个开放架构下的湖仓管理系统，在开放的数据湖格式之上，Arctic 提供更多面向流和更新场景的优化，以及一套可插拔的数据自优化机制和管理服务。基于 Arctic 可以帮助各类数据平台，工具和产品快速搭建开箱即用，流批统一的湖仓。
我们的湖仓目前使用的是 Hive catalog+Native Iceberg Format，接入大数据集群中的 Hive Metastore，以 Database 为单位配置需要 Arctic 管理的 Database。
Arctic 可以方便查看表及其详细信息，可以管理 Optimizer 服务，可以用标准命令行对表进行操作，通过 Arctic 可以很方便地对库表进行可视化展示和管理。
文件治理
数据入湖后，Arctic AMS 服务监听表的最新状态，根据阈值调度后台进行异步的合并小文件、清理孤儿文件和清理过期快照等任务。
在 Arctic 的架构中，这一部分被称作 Self-optimizing[7]，借鉴了 Java 虚拟机分代垃圾回收算法，将文件按照大小分为 Fragment 和 Segment，将 Fragment 和 Segment 上执行的不同 Self-optimizing 过程分为 Minor 和 Major 两种类型。很好地在保证读取性能的基础上，减小了写放大的影响。
Self-optimizing 的架构与工作机制如下图所示：
Optimizer 是 Self-optimizing 的执行组件，是由 AMS 管理的常驻进程，AMS 会负责发现和计划湖仓表的自优化任务，并实时调度给 Optimizer 分布式执行，最后由 AMS 负责提交优化结果，Arctic 通过 Optimizer Group 对 Optimizers 实现物理隔离。
Arctic 的 Self-optimizing 的核心特性有：
自动、异步与透明 — 后台持续检测文件变化，异步分布式执行优化任务，对用户透明无感 资源隔离和共享 — 允许资源在表级隔离和共享，以及设置资源配额 灵活可扩展的部署方式 — 执行节点支持多种部署方式，便捷的扩缩容
我们在实际生产中，配置的小文件合并阈值（self-optimizing.minor.trigger.file-count）为 12，也就是当小于 16m 的文件大于等于 12 个，会生成一个合并任务，然后通过 Optimizer 执行这个合并任务。在增量比较小的情况下，15min 左右会触发一次小文件合并。当然，可以通过参数调整来保证数据更加的实时。
开启孤儿文件自动清理，配置 clean-orphan-file.min-existing-time-minutes 为 2 天，AMS 会周期检查这些孤儿文件（任务或作业失败可能会留下表元数据未引用的文件），并删除它，避免存储的浪费。
现在 Iceberg 在企查查已经上线 200+ V2 表。最大单表为 1TB 左右，总的数据 4TB+。配置的 Optimizer 为 20 个并发，TaskManager 内存为 8G，slots 配置为 1，Optimizer 占用总的资源为 160G。Arctic 可以保持每张表的小文件数（＜16M）小于12，删除文件大小小于 64M 的状态。在实时更新业务场景下，普通查询能秒级响应，一般复杂查询分钟级响应。
社区贡献
Arctic 是一个非常 Open 的社区，在使用 Arctic 期间也积极参与社区贡献，截止目前已经在社区创建 15+ lssue，贡献了 10+ PR。其中包括 Self-optimizing 利用率优化，Self-optimizing 触发规则优化等多个重要贡献，使得 Self-optimizing 的性能以及效果得到了很大的提升。
目前社区也在进行 AMS 模块的重构，对整体代码结构及流程做了很大的优化，也会带来更好的任务调度策略和更高的资源利用率，我们也在积极参与。
未来规划
实时湖仓解决了此前数据服务的很多痛点，可以近实时数据分析，避免凌晨之后对源库抽取数据而产生大的影响，同时提升了 ODS 层的时效性，可以更早的调度后续的 ETL 任务，意味着可以更加快速的产出结果和导出到外部存储。
未来规划主要是下面几个方面：
持续优化 Arctic Optimizer 资源利用率和 Iceberg 的查询性能。 替换离线 ODS 层，预计 2000+ 表接入。 数据源种类比较多，支持更多的数据源接入。 接入对象存储，降低存储的成本，JuiceFS + MinIO
最后感谢 Apache Iceberg、Arctic 开源项目，特别鸣谢 Arctic 社区同学的帮助。
Reference
[1]: https://github.com/NetEase/arctic
[2]: https://www.dremio.com/blog/comparison-of-data-lake-table-formats-apache-iceberg-apache-hudi-and-delta-lake/
[3]:https://hudi.apache.org/docs/faq#what-options-do-i-have-for-asynchronousoffline-compactions-on-mor-dataset
[4]: B站 https://mp.weixin.qq.com/s/2arwvr6ityHwzFWNjd4hxQ
[5]: OPPO https://mp.weixin.qq.com/s/lPi99OMGH9fzjodrkh2Jdg
[6]: 小米 https://mp.weixin.qq.com/s/CZzw-ujxT6MdVho4KJ4sTA
[7]: https://arctic.netease.com/ch/concepts/self-optimizing/
END
看到这里 记得关注、点赞、转发 一键三连哦~
精彩回顾：
手把手教你使用 Arctic 自动优化 Apache Iceberg
关于 Arctic 的更多资讯可查看：
【GitHub】 地址：https://github.com/NetEase/arctic 【Arctic】 文档地址：https://arctic.netease.com/ch/
【社群】：扫描下方二维码加入 Arctic 社群↓ （也可以添加小助手微信：kllnn999 邀你进群）